package flume18;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Classname EventTSInterceptor
 * @Date 2020/11/10 20:48
 * @Create by device
 * @Desciption 自定义拦截器的工作机制，其实很简单：
 * flume先调自定义拦截器中的一个内部Builder类的config()方法进行参数配置
 * flume再调Builder类的build()方法获取自定义拦截器的实例对象(可以在构造过程中传递参数)
 * flume再反复调用拦截器对象的intercept(List<Event> events)方法来修改event
 */
public class EventTSInterceptor implements Interceptor {
    String timeStampFiledName;
    String toEncryFieldName;
    String keyName;
    Gson gson;

    public EventTSInterceptor(String timeStampFiledName,String toEncryFieldName,String keyName){
        this.timeStampFiledName = timeStampFiledName;
        this.toEncryFieldName = toEncryFieldName;
        this.keyName = keyName;
    }

    public void initialize() {
        gson = new Gson();
    }

    public Event intercept(Event event) {
        byte[] body = event.getBody();
        String lineJson = new String(body);
        //获取时间戳,放入header
        Map<String, Object> map = gson.fromJson(lineJson, new TypeToken<HashMap<String, Object>>() {
        }.getType());
        Double ts = (Double) map.get(timeStampFiledName);
        event.getHeaders().put(keyName, ts.longValue() + "");
        map.put("timeStamp", ts.longValue());
        // 将账号字段进行加密
        String s = (String) map.get(toEncryFieldName);
        String s1 = DigestUtils.md5Hex(s);
        map.put(toEncryFieldName, s1);
        // 将加密处理后的日志内容（map中），重新恢复成json，并set到event的body中
        String newJson = gson.toJson(map);
        event.setBody(newJson.getBytes());
        return event;
    }

    public List<Event> intercept(List<Event> list) {
        for (Event event : list) {
            intercept(event);

        }
        return list;
    }

    public void close() {

    }


    public static class EventTSInterceptorBuilder implements Interceptor.Builder {
        String timeStampFiledName;
        String toEncryFieldName;
        String keyName;

        public Interceptor build() {
            return new EventTSInterceptor(timeStampFiledName,toEncryFieldName,keyName);
        }

        public void configure(Context context) {
            timeStampFiledName = context.getString("tsFiledName");
            toEncryFieldName = context.getString("toEncryFieldName");
            keyName = context.getString("keyName");
        }
    }


}

